package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.junit.{Before, Test}

class Demo24Student {
  var stuRDD: RDD[Stu] = _
  var scoRDD: RDD[Sco] = _
  var subRDD: RDD[Sub] = _
  var sc: SparkContext = _

  def filterWithIdListAndPrint(ids: List[String]): Unit = {
    // 将ids进行广播
    val broIds: Broadcast[List[String]] = sc.broadcast(ids)

    // 将stuRDD变成KV格式 便于关联
    val stuKVRDD: RDD[(String, (String, String))] = stuRDD
      .filter(stu => broIds.value.contains(stu.id))
      .map(stu => (stu.id, (stu.name, stu.clazz)))

    // 将subRDD变成KV根式 便于关联
    val subKVRDD: RDD[(String, String)] = subRDD.map(sub => (sub.subId, sub.subName))

    // 学生表 关联 分数表
    scoRDD
      // 先过滤再关联
      .filter(sco => broIds.value.contains(sco.id))
      .map(sco => (sco.id, sco))
      .join(stuKVRDD)
      .map {
        case (id: String, (sco: Sco, (name: String, clazz: String))) =>
          (sco.subId, (id, name, clazz, sco.score))
      }
      .join(subKVRDD)
      .map {
        case (subId: String, ((id: String, name: String, clazz: String, score: Int), subName: String)) =>
          s"$id,$name,$clazz,$subName,$score"
      }
      .sortBy(s => s.split(",")(0)) // 按id排序结果方便查看数据
      .foreach(println)
  }

  @Before
  def init(): Unit = {
    // 读取三份数据 并构建对应的样例类对象 然后转换为RDD
    sc = new SparkContext(
      new SparkConf()
        .setAppName("Demo24Student")
        .setMaster("local")
    )

    stuRDD = sc.textFile("data/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Stu(splits(0), splits(1), splits(2).toInt, splits(3), splits(4))
      })

    scoRDD = sc.textFile("data/score.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Sco(splits(0), splits(1), splits(2).toInt)
      })

    subRDD = sc.textFile("data/subject.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Sub(splits(0), splits(1), splits(2).toInt)
      })

  }

  @Test
  def printRDD(): Unit = {
    stuRDD.take(10).foreach(println)
    scoRDD.take(10).foreach(println)
    subRDD.take(10).foreach(println)
  }

  @Test
  // 1、统计年级排名前十学生各科的分数 [学号,学生姓名，学生班级，科目名，分数]
  def question1(): Unit = {
    // 通过scoRDD计算学生总分 按降序排名 取前10的学生id 并关联学生、科目表
    val top10Ids: List[String] = scoRDD
      .map(sco => (sco.id, sco.score))
      .reduceByKey(_ + _) // 计算学生总分
      .sortBy(kv => kv._2, ascending = false) // 按照总分降序排列
      .map(kv => kv._1) // 不要总分 直接取id
      .take(10) // 取前十名
      .toList

    filterWithIdListAndPrint(top10Ids)

  }

  @Test
  // 2、统计总分大于年级平均分的学生 [学号，姓名，班级，总分]
  def question2(): Unit = {
    // 计算平均分 然后过滤出总分大于平均分的学生
    val sumScoreRDD: RDD[(String, Int)] = scoRDD
      .map(sco => (sco.id, sco.score))
      .reduceByKey(_ + _)

    // 对多次使用的RDD进行缓存
    sumScoreRDD.cache()

    val sumScoreAndCnt: (Int, Int) = sumScoreRDD
      .map(kv => (1, kv._2))
      .aggregateByKey((0, 0))(
        (u: (Int, Int), sumScor: Int) => (u._1 + sumScor, u._2 + 1),
        (u1, u2) => (u1._1 + u2._1, u1._2 + u2._2)
      ).collect()(0)._2

    // 平均成绩
    val avgSumScore: Double = sumScoreAndCnt._1.toDouble / sumScoreAndCnt._2

    println(avgSumScore)

    // 过滤出总分大于平均分的学生
    val passSumScoreRDD: RDD[(String, Int)] = sumScoreRDD
      .filter(kv => kv._2 > avgSumScore)

    passSumScoreRDD.cache()

    // 取出总分大于平均分的学生的id
    val passIDs: List[String] = passSumScoreRDD.map(kv => kv._1).collect().toList

    val broPassIDs: Broadcast[List[String]] = sc.broadcast(passIDs)

    // 将stuRDD变成KV格式 便于关联
    val stuKVRDD: RDD[(String, (String, String))] = stuRDD
      .filter(stu => broPassIDs.value.contains(stu.id))
      .map(stu => (stu.id, (stu.name, stu.clazz)))

    passSumScoreRDD
      .join(stuKVRDD)
      .map {
        case (id: String, (sumScore: Int, (name: String, clazz: String))) =>
          s"$id,$name,$clazz,$sumScore"
      }.foreach(println)

    // 记得释放缓存
    passSumScoreRDD.unpersist()
    sumScoreRDD.unpersist()
  }

  @Test
  // 3、统计每科都及格的学生 [学号，姓名，班级，科目，分数]
  def question3(): Unit = {
    // 将每个学生不及格的科目成绩记录去除
    val subKVRDD: RDD[(String, Int)] = subRDD
      .map(sub => (sub.subId, sub.subScore))

    // 找到所有科目都及格的学生的id
    val passAllSubIds: List[String] = scoRDD
      .map(sco => (sco.subId, sco))
      .join(subKVRDD)
      .filter {
        case (subId: String, (sco: Sco, subScore: Int)) =>
          sco.score >= subScore * 0.6
      }
      .map {
        case (subId: String, (sco: Sco, subScore: Int)) =>
          (sco.id, 1)
      }
      .reduceByKey(_ + _) // 统计每个学生及格的科目数量
      .filter(kv => kv._2 == 6) // 取出6门科目都及格的学生成绩
      .map(_._1) // 将id取出
      .collect()
      .toList

    filterWithIdListAndPrint(passAllSubIds)

  }

  @Test
  // 4、统计每个班级的前三名 [学号，姓名，班级，分数]
  def question4(): Unit = {
    // 将stuRDD变成KV格式 便于关联
    val stuKVRDD: RDD[(String, (String, String))] = stuRDD
      .map(stu => (stu.id, (stu.name, stu.clazz)))


    scoRDD
      .map(sco => (sco.id, sco.score))
      .reduceByKey(_ + _) // 计算学生总分
      .join(stuKVRDD)
      .map {
        case (id: String, (sumScore: Int, (name: String, clazz: String))) =>
          (id, name, clazz, sumScore)
      }
      .groupBy(t4 => t4._3)
      .flatMap {
        case (clazz: String, t4: Iterable[(String, String, String, Int)]) =>
          // 取出班级前三名
          t4.toList.sortBy(t4 => -t4._4).take(3)
      }
      .map(t4 => s"${t4._1},${t4._2},${t4._3},${t4._4}")
      .foreach(println)

  }

  @Test
  // 5、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数]
  def question5(): Unit = {
    // 找到偏科最严重的前100名学生的id

    // 归一化 --> 方差 --> 排序 -> 取前100名

    // 将每个学生不及格的科目成绩记录去除
    val subKVRDD: RDD[(String, Int)] = subRDD
      .map(sub => (sub.subId, sub.subScore))

    val top100ids: List[String] = scoRDD
      .map(sco => (sco.subId, sco))
      .join(subKVRDD)
      .map {
        case (subId: String, (sco: Sco, subScore: Int)) =>
          (sco.id, sco.score * 100.0 / subScore)
      }
      .groupByKey()
      .map(kv => {
        val id: String = kv._1
        val scores: Iterable[Double] = kv._2
        val avgScore: Double = scores.sum / scores.size
        val variance: Double = scores
          .map(score => {
            Math.pow(score - avgScore, 2)
          }).sum / scores.size
        (id, variance)
      })
      .sortBy(-_._2)
      .map(_._1)
      .take(100)
      .toList

    filterWithIdListAndPrint(top100ids)

  }


}

case class Stu(id: String, name: String, age: Int, gender: String, clazz: String)

case class Sco(id: String, subId: String, score: Int)

case class Sub(subId: String, subName: String, subScore: Int)
